package utils

import (
	"sync"
	"sync/atomic"
	"time"
)

/*
Package queue provides a fast, ring-buffer queue based on the version suggested by Dariusz Górecki.
Using this instead of other, simpler, queue implementations (slice+append or linked list) provides
substantial memory and time benefits, and fewer GC pauses.
The queue implemented here is as fast as it is for an additional reason: it is *not* thread-safe.
*/

// minQueueLen is the smallest capacity that queue may have.
// Must be power of 2 for bitwise modulus: x % n == x & (n - 1).
const minQueueLen = 1024

// BlockQueue represents a single instance of the queue data structure.
// Chan using for notice consumer
type BlockQueue struct {
	Buf     [][]byte
	readBuf [][]byte
	count   *uint32
	HasData chan bool
	cond    *sync.Cond
	mutex   *sync.RWMutex
	seq     uint32
	pre     time.Time
	minPoll int64
	size    uint32
}

// NewDefaultQueue constructs and returns a new BlockQueue.
func NewDefaultQueue() *BlockQueue {
	lock := new(sync.RWMutex)
	return &BlockQueue{
		Buf:     make([][]byte, minQueueLen),
		readBuf: make([][]byte, minQueueLen),
		count:   new(uint32),
		HasData: make(chan bool, 1),
		cond:    sync.NewCond(lock.RLocker()),
		mutex:   lock,
		seq:     uint32(1),
		pre:     time.Now(),
		minPoll: int64(time.Nanosecond * 50),
		size:    minQueueLen,
	}
}

// NewQueue constructs and returns a new BlockQueue.
func NewQueue(size int, minPoll time.Duration) *BlockQueue {
	lock := new(sync.RWMutex)
	return &BlockQueue{
		Buf:     make([][]byte, size),
		readBuf: make([][]byte, size),
		count:   new(uint32),
		HasData: make(chan bool, 1),
		cond:    sync.NewCond(lock),
		mutex:   lock,
		seq:     uint32(1),
		pre:     time.Now(),
		minPoll: int64(minPoll),
		size:    uint32(size),
	}
}

// Length returns the number of elements currently stored in the queue.
func (q *BlockQueue) Length() uint32 {
	return atomic.LoadUint32(q.count)
}

// Add puts an element on the end of the queue.
func (q *BlockQueue) Add(elem []byte) uint32 {
	q.cond.L.Lock()
	defer q.cond.L.Unlock()
	for atomic.LoadUint32(q.count) == q.size {
		q.cond.Wait()
	}
	val := atomic.LoadUint32(q.count)
	if val == 0 {
		q.HasData <- true
	}
	q.Buf[val] = elem
	atomic.AddUint32(q.count, 1)
	q.cond.Broadcast()
	return q.seq
}

func (q *BlockQueue) PollAll() ([][]byte, uint32) {
	<-q.HasData
	delta := time.Now().Sub(q.pre)
	if delta.Nanoseconds() < q.minPoll {
		time.Sleep(delta)
		if q.minPoll > 1 {
			q.minPoll--
		}
	} else {
		if q.minPoll < 10000 {
			q.minPoll++
		}
	}
	v, s := q.LoadAll()
	q.pre = time.Now()
	return v, s
}

func (q *BlockQueue) LoadAll() ([][]byte, uint32) {
	q.mutex.Lock()
	defer q.mutex.Unlock()
	val := atomic.LoadUint32(q.count)
	copy(q.readBuf[:val], q.Buf[:val])
	atomic.StoreUint32(q.count, 0)
	q.seq = q.seq + 1
	if val == q.size {
		q.cond.Broadcast()
	}
	// notice empty
	return q.readBuf[:val], q.seq
}
